/*
 * Copyright (c) 2011 Imre Fazekas.
 * All rights reserved.
 * Redistribution and use in source and binary forms, with or without
 * modification, are permitted provided that the following conditions are met:
 *
 * Redistributions of source code must retain the above copyright notice, this
 * list of conditions and the following disclaimer.
 *
 * Redistributions in binary form must reproduce the above copyright notice,
 * this list of conditions and the following disclaimer in the documentation
 * and/or other materials provided with the distribution.
 * Neither the name of the Brillien nor the names of its
 * terms and concepts may be used to endorse or promote products derived from this
 * software without specific prior written permission.
 *
 * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
 * AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
 * IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
 * ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER OR CONTRIBUTORS BE
 * LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR
 * CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF
 * SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS
 * INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN
 * CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE)
 * ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
 * POSSIBILITY OF SUCH DAMAGE.
 */

package com.vii.streamline.web.providers;

import com.vii.streamline.json.JsonRpcRequest;
import com.vii.streamline.services.db.transaction.BrillienTransaction;
import com.vii.streamline.services.error.StreamLineException;
import com.vii.streamline.services.json.JsonServices;
import com.vii.streamline.web.DBDataProvider;

import javax.servlet.ServletConfig;
import java.sql.*;
import java.util.Collection;
import java.util.HashMap;
import java.util.Map;

/**
 * Specialized DBDataProvider managing PostgreSQL database
 */
public class PgProvider extends DBDataProvider {

    private static final String CREATE_QUEUE_TABLE = "CREATE TABLE ?\n" +
            "(\n" +
            "  queue_id serial NOT NULL,\n" +
            "  message text NOT NULL,\n" +
            "  received_date timestamp with time zone NOT NULL,\n" +
            "  CONSTRAINT ?_pk PRIMARY KEY (queue_id)\n" +
            ")\n" +
            "WITH (\n" +
            "  OIDS=FALSE\n" +
            ");";

    public static final String IS_QUEUE_EXISTS = "SELECT count(*) FROM ?";

    public static final String INSERT_QUEUE_TABLE = "INSERT INTO ?\n" +
            "(message, received_date)\n" +
            "VALUES\n" +
            "(?, ?) RETURNING queue_id as id;";

    public static final String GET_LATEST_MESSAGE = "SELECT queue_id, message, received_date FROM ?\n" +
            "WHERE received_date > (now() - interval '? second' );";

    public static final String REMOVE_MESSAGE = "DELETE FROM ?\n" +
            "WHERE queue_id = ?;";

    public PgProvider(ServletConfig config) throws StreamLineException {
        super(config);
    }

    public PgProvider(String vendor, String username, String password, String address, String port, String database) throws StreamLineException {
        super(vendor, username, password, address, port, database);
    }

    protected boolean isQueueTableExists( String queue ) throws SQLException {
        Connection c        = null;
        PreparedStatement p = null;
        try{
            c = getDs().getConnection();
            p = c.prepareStatement( IS_QUEUE_EXISTS.replace("?", queue) );

            p.executeQuery();

            return true;
        } catch(Exception e){ }
        finally{
            if( p != null )
                p.close();

            if( c != null )
                c.close();
        }
        return false;
    }

    protected void createQueueTable( String queue ) throws SQLException {
        Connection c        = null;
        PreparedStatement p = null;
        try{
            c = getDs().getConnection();
            p = c.prepareStatement( CREATE_QUEUE_TABLE.replaceAll("\\?", queue) );

            System.out.println("" + p.toString() );

            p.executeUpdate();
        }
        finally{
            if( p != null )
                p.close();

            if( c != null )
                c.close();
        }
    }

    protected void insertMessage(String queue, String message) throws SQLException, StreamLineException {
        Connection c        = null;
        PreparedStatement p = null;
        try{
            c = getConnection( true );
            p = c.prepareStatement( INSERT_QUEUE_TABLE.replaceFirst("\\?", queue) );

            p.setString(1, message);
            p.setTimestamp(2, new Timestamp(System.currentTimeMillis()));

            p.executeQuery();
        }
        finally{
            if( p != null )
                p.close();

            if( c != null )
                retrieveConnection( c );
        }
    }

    @Override
    public void store(String recipient, JsonRpcRequest request) throws StreamLineException {
        try{
            if( !isQueueTableExists( recipient ) ){
                createQueueTable( recipient );
            }
            insertMessage( recipient, JsonServices.toJSON( request ) );
        } catch (SQLException e) {
            throw new StreamLineException( e );
        }
    }

    @Override
    public void remove(String queuename, Long requestId) throws StreamLineException{
        Connection c        = null;
        PreparedStatement p = null;
        try{
            c = getConnection( true );
            p = c.prepareStatement( REMOVE_MESSAGE );

            p.setString(1, queuename);
            p.setLong(2, requestId);
            p.executeUpdate();

        } catch (SQLException e) {
            throw new StreamLineException( e );
        } finally{
            if( p != null )
                try {
                    p.close();
                } catch (SQLException e) { }

            if( c != null )
                try {
                    retrieveConnection( c );
                } catch (StreamLineException e) { }
        }
    }

    @Override
    public void remove(String queuename, Collection<Long> requestId) throws StreamLineException{
        Connection c        = null;
        PreparedStatement p = null;
        try{
            BrillienTransaction t = startTransaction();

            c = getConnection(false);
            t.enlistConnection( c );

            p = c.prepareStatement( REMOVE_MESSAGE );
            for( Long id : requestId ){
                p.setString( 1, queuename );
                p.setLong( 2, id );
                p.executeUpdate();
                p.clearParameters();
            }

            retrieveConnection(c);
        } catch (Exception e) {
            throw new StreamLineException( e );
        } finally{
            if( p != null )
                try {
                    p.close();
                } catch (SQLException e) { }

            if( c != null )
                try {
                    c.close();
                } catch (SQLException e) { }
        }
    }


    @Override
    public Map<Long,JsonRpcRequest> getUndeliveredRequests(String queuename, long expirationTime) throws StreamLineException {
        HashMap<Long,JsonRpcRequest> requests = new HashMap<Long,JsonRpcRequest>();
        Connection c        = null;
        PreparedStatement p = null;
        try{
            c = getConnection( true );
            p = c.prepareStatement( GET_LATEST_MESSAGE
                .replaceFirst("\\?", queuename)
                .replaceFirst("\\?", (expirationTime / 1000) + "")
            );

            ResultSet r = p.executeQuery();
            while( r.next() ){
                requests.put(
                    r.getLong("queue_id"),
                    JsonServices.<JsonRpcRequest>parseJSON(r.getString("message"), JsonRpcRequest.class)
                );
            }
        } catch (SQLException e) {
            throw new StreamLineException( e );
        } finally{
            if( p != null )
                try {
                    p.close();
                } catch (SQLException e) { }

            if( c != null )
                try {
                    retrieveConnection( c );
                } catch (StreamLineException e) { }
        }
        return requests;
    }

}
